package com.ldj.rocketmq.producer;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

import java.nio.charset.StandardCharsets;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * User: ldj
 * Date: 2024/3/26
 * Time: 2:05
 * Description: 异步消息生产者 使用场景：不容忍主业务长等待的场景
 */
public class AsyncProducer {

    public static void main(String[] args) {

        DefaultMQProducer producer = new DefaultMQProducer("produce-group1");
        producer.setNamesrvAddr("192.168.208.190:9876;192.168.208.191:9876;92.168.208.192:9876");
        try {
            producer.start();
            for (int i = 0; i < 10; i++) {
                String msg = StringUtils.join(" I've got an appointment in fifteen minutes ", i + 1);
                Message message = new Message("love_letter", "appointment ", msg.getBytes(StandardCharsets.UTF_8));
                //异步消息比同步信息多一个发送后的回调函数
                producer.send(message, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult result) {
                        Map<String, Object> map = new LinkedHashMap<>();
                        map.put("Broker名称", result.getMessageQueue().getBrokerName());
                        map.put("消息队列Id", result.getMessageQueue().getQueueId());
                        map.put("消息Id", result.getMsgId());
                        map.put("消息状态", result.getSendStatus());
                        System.out.printf("消息发送成功！%s", map);
                    }

                    @Override
                    public void onException(Throwable e) {
                        System.out.printf("消息发送失败！%s", e.getMessage());
                    }
                }, 3 * 1000);

                TimeUnit.SECONDS.sleep(1);
            }
        } catch (Exception e) {
            System.out.println(e.getMessage());
        } finally {
            producer.shutdown();
        }
    }
}
